package cn.dmp.tools

import org.apache.spark.rdd.RDD
import org.apache.spark.sql.types._
import org.apache.spark.sql.{Row, SQLContext}
import org.apache.spark.{SparkConf, SparkContext}

/**
  * Created by Administrator on 2018/4/18.
  */
object Biz2Parquet {
  def main(args: Array[String]): Unit = {

    if (args.length != 2){
      println(
        """
          |cd.dmp.tool.Biz2Parquet
          |参数不合法，请输入正确参数个数：dataInputPath、dataOutPath
        """.stripMargin)
      sys.exit(1)
    }
    val conf = new SparkConf()
      .setMaster("local[*]").setAppName("Biz2Parquet")
      .set("spark.serializer","org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)  //spark离线job的入口

    //输入参数 （模糊匹配）
    val Array(dataInputPath,dataOutPath)=args
    //读取数据
    val data: RDD[String] = sc.textFile(dataInputPath)

    //切分数据（-1是指读到行尾）并过滤  数据的长度大于85
    val filter: RDD[Array[String]] = data.map(_.split(",",-1)).filter(t=>t.length>=85)

    import cn.dmp.beans.RichString._

    val rddRowMap: RDD[Row] = filter.map(t => Row(
      t(0),
      t(1).String2Int,
      t(2).String2Int,
      t(3).String2Int,
      t(4).String2Int,
      t(5),
      t(6),
      t(7).String2Int,
      t(8).String2Int,
      t(9).String2Double,
      t(10).String2Double,
      t(11),
      t(12),
      t(13),
      t(14),
      t(15),
      t(16),
      t(17).String2Int,
      t(18),
      t(19),
      t(20).String2Int,
      t(21).String2Int,
      t(22),
      t(23),
      t(24),
      t(25),
      t(26).String2Int,
      t(27),
      t(28).String2Int,
      t(29),
      t(30).String2Int,
      t(31).String2Int,
      t(32).String2Int,
      t(33),
      t(34).String2Int,
      t(35).String2Int,
      t(36).String2Int,
      t(37),
      t(38).String2Int,
      t(39).String2Int,
      t(40).String2Double,
      t(41).String2Double,
      t(42).String2Int,
      t(43),
      t(44).String2Double,
      t(45).String2Double,
      t(46),
      t(47),
      t(48),
      t(49),
      t(50),
      t(51),
      t(52),
      t(53),
      t(54),
      t(55),
      t(56),
      t(57).String2Int,
      t(58).String2Double,
      t(59).String2Int,
      t(60).String2Int,
      t(61),
      t(62),
      t(63),
      t(64),
      t(65),
      t(66),
      t(67),
      t(68),
      t(69),
      t(70),
      t(71),
      t(72),
      t(73).String2Int,
      t(74).String2Double,
      t(75).String2Double,
      t(76).String2Double,
      t(77).String2Double,
      t(78).String2Double,
      t(79),
      t(80),
      t(81),
      t(82),
      t(83),
      t(84).String2Int
    ))

    //定义schema信息
    val structType = StructType(List(
      StructField("sessionid", StringType),
      StructField("advertisersid", IntegerType),
      StructField("adorderid", IntegerType),
      StructField("adcreativeid", IntegerType),
      StructField("adplatformproviderid", IntegerType),
      StructField("sdkversion", StringType),
      StructField("adplatformkey", StringType),
      StructField("putinmodeltype", IntegerType),
      StructField("requestmode", IntegerType),
      StructField("adprice", DoubleType),
      StructField("adppprice", DoubleType),
      StructField("requestdate", StringType),
      StructField("ip", StringType),
      StructField("appid", StringType),
      StructField("appname", StringType),
      StructField("uuid", StringType),
      StructField("device", StringType),
      StructField("client", IntegerType),
      StructField("osversion", StringType),
      StructField("density", StringType),
      StructField("pw", IntegerType),
      StructField("ph", IntegerType),
      StructField("long", StringType),
      StructField("lat", StringType),
      StructField("provincename", StringType),
      StructField("cityname", StringType),
      StructField("ispid", IntegerType),
      StructField("ispname", StringType),
      StructField("networkmannerid", IntegerType),
      StructField("networkmannername", StringType),
      StructField("iseffective", IntegerType),
      StructField("isbilling", IntegerType),
      StructField("adspacetype", IntegerType),
      StructField("adspacetypename", StringType),
      StructField("devicetype", IntegerType),
      StructField("processnode", IntegerType),
      StructField("apptype", IntegerType),
      StructField("district", StringType),
      StructField("paymode", IntegerType),
      StructField("isbid", IntegerType),
      StructField("bidprice", DoubleType),
      StructField("winprice", DoubleType),
      StructField("iswin", IntegerType),
      StructField("cur", StringType),
      StructField("rate", DoubleType),
      StructField("cnywinprice", DoubleType),
      StructField("imei", StringType),
      StructField("mac", StringType),
      StructField("idfa", StringType),
      StructField("openudid", StringType),
      StructField("androidid", StringType),
      StructField("rtbprovince", StringType),
      StructField("rtbcity", StringType),
      StructField("rtbdistrict", StringType),
      StructField("rtbstreet", StringType),
      StructField("storeurl", StringType),
      StructField("realip", StringType),
      StructField("isqualityapp", IntegerType),
      StructField("bidfloor", DoubleType),
      StructField("aw", IntegerType),
      StructField("ah", IntegerType),
      StructField("imeimd5", StringType),
      StructField("macmd5", StringType),
      StructField("idfamd5", StringType),
      StructField("openudidmd5", StringType),
      StructField("androididmd5", StringType),
      StructField("imeisha1", StringType),
      StructField("macsha1", StringType),
      StructField("idfasha1", StringType),
      StructField("openudidsha1", StringType),
      StructField("androididsha1", StringType),
      StructField("uuidunknow", StringType),
      StructField("userid", StringType),
      StructField("iptype", IntegerType),
      StructField("initbidprice", DoubleType),
      StructField("adpayment", DoubleType),
      StructField("agentrate", DoubleType),
      StructField("lomarkrate", DoubleType),
      StructField("adxrate", DoubleType),
      StructField("title", StringType),
      StructField("keywords", StringType),
      StructField("tagid", StringType),
      StructField("callbackdate", StringType),
      StructField("channelid", StringType),
      StructField("mediatype", IntegerType)
    ))

    val sqlContext = new SQLContext(sc)
    //设置parquet文件的压缩格式
    sqlContext.setConf("spark.sql.parquet.compression.codec","snappy")
    //创建dataFram
    val dataFrame = sqlContext.createDataFrame(rddRowMap,structType)

    //dataFrame.write.parquet(dataOutPath)

    //分区存储,省和市是层级的关系，省是1个文件夹下面的市是一个文件夹
    dataFrame.write.partitionBy("provincename","cityname").parquet(dataOutPath)

    sc.stop()
  }
}
